package com.sunzm.flink.datastream.scala.state

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 *
 * ScalaStateDemo
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-06-22 20:48
 */
object ScalaStateDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {

    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    val dataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)

    val keyedDataStream: KeyedStream[(String, Int), String] = dataStream.flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)

    val wordCountDS: DataStream[(String, Int)] = keyedDataStream.flatMap(new RichFlatMapFunction[((String, Int)), (String, Int)] {

      /**
       * 记录每个单词次数的状态
       */
      private var countState: ValueState[Int] = _

      /**
       * 程序启动的时候，这个方法只会执行一次
       *
       * @param parameters
       */
      override def open(parameters: Configuration): Unit = {
        countState = getRuntimeContext.getState(
          //new ValueStateDescriptor[Int]("sum", classOf[Int])
          new ValueStateDescriptor[Int]("wordCountState", createTypeInformation[Int])
        )
      }

      /**
       * 每收到一条数据，就会执行一次
       *
       * @param wordAndOne
       * @param collector
       */
      override def flatMap(wordAndOne: (String, Int), collector: Collector[(String, Int)]): Unit = {

        // 获取状态数据中保存的当前这个单词的次数
        val tmpCurrentCout = countState.value

        // 如果是第一次使用状态，它的值可能是空，所以这里做一下处理
        //获取到当前状态中真实存储的单词次数
        val currentCount = if (tmpCurrentCout != null) {
          tmpCurrentCout
        } else {
          0
        }

        val one = wordAndOne._2
        val newCount = one + currentCount
        // 计算新的单词次数
        val word = wordAndOne._1

        // 更新状态
        countState.update(newCount)

        //将计算得到的最新的单词和次数输出
        collector.collect((word, newCount))

      }

      /**
       * 程序关闭的时候，会执行一次
       */
      override def close(): Unit = {
        //程序退出不能清空状态，不然重启后状态数据就无法恢复了
      }
    })


    wordCountDS.print()

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
